#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <MQTTClient.h>
#include "cJSON.h"
#include <unistd.h>
#include <pthread.h>
// #include "mysqlite3.h"
#include <sys/shm.h>
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>

typedef struct 
{
    long mstype;
    char mstext[256];
}MSG;
#define MSGLEN  (sizeof(MSG)-sizeof(long)) 

#define ADDRESS     "tcp://localhost:1883"  // MQTT代理地址
#define CLIENTID    "ExampleClientPub"    // 客户端ID
#define TOPICUP       "MQTT up"         // 主题
#define TOPICDOWN       "MQTT down"         // 主题
// #define PAYLOAD     "Hello World!"        // 消息内容
#define QOS         1
#define TIMEOUT     10000L

void mq_send(char *p);
char buf[2] = "";
int flg = 0;
int msgid;
MQTTClient client;
MQTTClient_message pubmsg = MQTTClient_message_initializer;
MQTTClient_deliveryToken token;
void delivered(void *context, MQTTClient_deliveryToken dt) {
    printf("Message with token value %d delivery confirmed\n", dt);
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message) 
{
    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: %.*s\n", message->payloadlen, (char*)message->payload);
    cJSON* json_qt = NULL;
  
       /* 解析整段JSO数据 */
    json_qt = cJSON_Parse((char*)message->payload);
    if(json_qt == NULL)
    {
        printf("parse fail.\n");
        return -1;
    }
    flg = 1;
    cJSON *type = NULL;
    type =  cJSON_GetObjectItem(json_qt, "type");
    if(type != NULL )
    {
        FILE * fp;
        fp = fopen("./node.txt","w");
        if(fp == NULL)
        {
            perror("fopen err");
            return -1;
        }
        fwrite(type->valuestring, strlen(type->valuestring), 1,fp);

        fclose(fp);
    }
    cJSON_Delete(json_qt);
    mq_send((char*)message->payload);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}
void mq_send(char *p)
{

    //编辑消息队列信息
    MSG msgT;
    memset(&msgT,0,sizeof(msgT));
    //不同进程通过消息队列获取信息的一依据
    msgT.mstype = 20;
    strcpy(msgT.mstext, p);
    //strcpy(msgT.mstext,"bob:hello msg");
 
    //发送消息
    msgsnd(msgid,&msgT,MSGLEN,0);
    printf("消息发送成功!\n");
}

void connlost(void *context, char *cause) {
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}


   void mqtt_send(char *p)
    {
		printf("%s\n", p);
        pubmsg.qos = QOS;
        pubmsg.retained = 0;
        pubmsg.payload = p;
        pubmsg.payloadlen = (int)strlen(p);
        MQTTClient_publishMessage(client, TOPICUP, &pubmsg, &token);
    }
// void sendAllData(char *p)
// {
//     char *p = cJSON_Print(root);
//     mqtt_send(p);
//     printf("上报\n");
//     free(p);
//     cJSON_Delete(root);
// }

//对比上报
void *up_vary(void* p)
{
    char new[64];
    // mqtt_send( p);
    if (strcmp(new,p) != 0)
    {
        mqtt_send( p);
        strcpy(new, p);
     }
}
//循环商报
void *up_period(void *p) 
{
    mqtt_send( p);
}

 //刷新上报
 void *up_refresh(void* p)
 {
    if(flg == 1)
    {
        mqtt_send( p);
     }
    flg = 0;

 }

int main(int argc, char* argv[])
 {

    //读取本地配置
    FILE * fp;
    fp = fopen("./node.txt","r");
    if(fp == NULL)
    {
        perror("fopen err");
        return -1;
    }
    fread(buf,1,1,fp);
    fclose(fp);
    printf("%s\n",buf);

    // //开启数据库
    // sqlite3 *db = NULL;
    // db = create_dictol_db("test.db");
    // create_table_db(db);
    
    //多线程标识符
    pthread_t tid;
    //创建消息队列
    msgid = msgget(0x300, IPC_CREAT | 0666 );
    printf("消息队列标识符为%d\n",msgid);
    if(msgid < 0)
	{
		perror("msgget");
		return -1;
	}
    //共享内存
	int  shm_id;
    //创建共享内存
	shm_id = shmget(0x300, 100, IPC_CREAT|0777);
	if(shm_id < 0)
	{
		perror("shmget");
		return -1;
	}
    //获取共享内存地址
	char *p  = shmat(shm_id, NULL, 0);

    

    MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
    int rc;

    MQTTClient_create(&client, ADDRESS, CLIENTID, MQTTCLIENT_PERSISTENCE_NONE, NULL);
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;

    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS) {
        printf("Failed to connect, return code %s\n", TOPICUP);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, TOPICDOWN, QOS);
    MQTTClient_publishMessage(client, TOPICUP, &pubmsg, &token);
    rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);

        
    // 等待接收消息，此处为阻塞等待，可以通过其他方式实现非阻塞或定时等待
    while (1) 
    {
        FILE * fp;
        fp = fopen("./node.txt","r");
        if(fp == NULL)
        {
            perror("fopen err");
            return -1;
        }
        fread(buf,1,1,fp);
        fclose(fp);
            
        if(buf[0] == '0')//不上报
        {
            printf("当前上报模式：不上报\n");
        }else if(buf[0] == '1')//循环上报
        {
            printf("当前上报模式：循环上报\n");
            up_period(p);
            // int ret = pthread_create(&tid, NULL, up_period, (void*)p);
            // if (ret != 0) 
            // {  
            //     printf("Failed to create thread\n");  
            //     return 1;  
            // }
        }
        else if(buf[0] =='2')//刷新上报
        {
            printf("当前上报模式：刷新上报\n");
            up_refresh(p);
            // int ret = pthread_create(&tid, NULL, up_refresh, (void*)p);
            // if (ret != 0) 
            // {  
            //     printf("Failed to create thread\n");  
            //     return 1;  
            // }
        }
        else if(buf[0] =='3')//对比上报
        {
            printf("当前上报模式：对比上报\n");
            up_vary(p);
            // int ret = pthread_create(&tid, NULL,up_vary, &p);
            // if (ret != 0) 
            // {  
            //     printf("Failed to create thread\n");  
            //     return 1;  
            // }
        }
    }

    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}
